package com.canal.test;

import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.protocol.CanalEntry;
import com.alibaba.otter.canal.protocol.Message;
import com.canal.util.CanalUtils;

import java.util.List;

public class ClientSample {
  public static void main(String args[]) {
    CanalConnector canalConnector = CanalUtils.getSingleCanalConnector("101.200.168.222", 11111, "test-instance", "", "");
    //计数器
    int emptyCount = 0;
    //一次最多拉多少条Message
    //注意:

    int batchSize = 1000;
    try {
      //和server建立连接
      canalConnector.connect();
      //订阅表
      canalConnector.subscribe("canal.test_source");
      //回滚到上次ack的位置
      canalConnector.rollback();
      //最大空闲次数
      int maxEmptyCount = 1000;
      //死循环去拉取数据
      while (emptyCount < maxEmptyCount) {
        //尝试最多拿batchSize条记录
        //注意:getWithoutAck一次，对应一个Message。
        //一个Message有一个MessageID,还有一个List<CanalEntry.Entry>
        Message message = canalConnector.getWithoutAck(batchSize);
        long batchId = message.getId();
        int size = message.getEntries().size();
        //没有数据---等待
        if (batchId == -1 || size == 0) {
          emptyCount++;
          //                    System.out.println("empty count : " + emptyCount);
          try {
            Thread.sleep(1000);
          } catch (InterruptedException e) {
            e.printStackTrace();
          }
        } else {
          //拿到数据---开始解析--发送到目的地如kafka/elasticsearch/redis/hbase...等
          emptyCount = 0;
          //这里打印内容
          printEntry(message.getEntries());
        }

        canalConnector.ack(batchId); // 提交确认
        //canalConnector.rollback(batchId); // 处理失败, 回滚数据
      }

      System.out.println("empty too many times, exit");
    } finally {
      canalConnector.disconnect();
    }
  }

  private static void printEntry(List<CanalEntry.Entry> entrys) {
    for (CanalEntry.Entry entry : entrys) {
      //该条数据的数据类型是事务开始或事务结束，不是binlog 二进制数据本身，就跳过继续处理
      //注意:这里是有bug的，当mysql开启binlog，且为Row行模式，且开启了在binlog中显示原始SQL。
      //这时就会有一种新增的类型:CanalEntry.EventType.QUERY
      // 原始
      //if (entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONBEGIN || entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONEND) {
      //    continue;
      //}
      // 修改为 只保留binlog部分对应的Entry并解析
      if (entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONBEGIN //跳过事务开始的Entry
          || entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONEND //跳过事务结束的Entry
          || entry.getHeader().getEventType() == CanalEntry.EventType.QUERY //跳过事务为原始SQL的Entry
      ) {
        continue;
      }
      //得到当前行变化的数据Before、After
      CanalEntry.RowChange rowChage = null;
      try {
        rowChage = CanalEntry.RowChange.parseFrom(entry.getStoreValue());
      } catch (Exception e) {
        throw new RuntimeException("ERROR ## parser of eromanga-event has an error , data:" + entry.toString(), e);
      }

      //事件类型
      CanalEntry.EventType eventType = rowChage.getEventType();
      System.out.println(String.format("================> binlog[%s:%s] , name[%s,%s] , eventType : %s", entry.getHeader().getLogfileName(), //binlog文件名
          entry.getHeader().getLogfileOffset(),//binlog offset
          entry.getHeader().getSchemaName(),//库名
          entry.getHeader().getTableName(),//表名
          eventType //事件类型
      ));

      for (CanalEntry.RowData rowData : rowChage.getRowDatasList()) {
        //INSERT
        if (eventType == CanalEntry.EventType.INSERT) {
          printColumn(rowData.getAfterColumnsList());
          //DELETE
        } else if (eventType == CanalEntry.EventType.DELETE) {
          printColumn(rowData.getBeforeColumnsList());
          //UPDATE
        } else if (eventType == CanalEntry.EventType.UPDATE) {
          System.out.println("-------> before");
          printColumn(rowData.getBeforeColumnsList());
          System.out.println("-------> after");
          printColumn(rowData.getAfterColumnsList());
        }
      }
    }
  }

  private static void printColumn(List<CanalEntry.Column> columns) {
    for (CanalEntry.Column column : columns) {
      //列名:列值 该列是否被更新
      System.out.println(column.getName() + " : " + column.getValue() + "    update=" + column.getUpdated());
    }
  }

}